package com.fanxuankai.boot.canal.client;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import com.fanxuankai.boot.canal.client.autoconfigure.ConsumerConfig;
import com.fanxuankai.boot.canal.util.CommonUtils;
import com.fanxuankai.boot.canal.util.Topic;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author fanxuankai
 */
public class MessageInfo {
    private String topic;
    private List<String> messages;

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public List<String> getMessages() {
        return messages;
    }

    public void setMessages(List<String> messages) {
        this.messages = messages;
    }

    public MessageInfo(EntryWrapper entryWrapper, CanalClientProperties properties) {
        String schemaName = entryWrapper.getSchemaName();
        String tableName = entryWrapper.getTableName();
        ConsumerConfig consumerConfig = properties.getConsumerConfig(entryWrapper).orElse(null);
        setTopic(Topic.of(entryWrapper, properties));
        setMessages(entryWrapper.getAllRowDataList()
                .stream()
                .map(rowData -> {
                    if (entryWrapper.getEventType() == CanalEntry.EventType.INSERT) {
                        return CommonUtils.jsonWithActualType(consumerConfig,
                                rowData.getAfterColumnsList(), schemaName,
                                tableName, false);
                    } else if (entryWrapper.getEventType() == CanalEntry.EventType.UPDATE) {
                        return CommonUtils.jsonWithActualType(consumerConfig,
                                rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(),
                                schemaName, tableName, false);
                    } else if (entryWrapper.getEventType() == CanalEntry.EventType.DELETE) {
                        return CommonUtils.jsonWithActualType(consumerConfig,
                                rowData.getBeforeColumnsList(), schemaName,
                                tableName, false);
                    }
                    return null;
                }).filter(Objects::nonNull)
                .collect(Collectors.toList()));
    }
}
